package com.qkjia.producer;

import com.alibaba.fastjson.JSONObject;
import com.qkjia.entity.Device;
import java.util.UUID;
import javax.annotation.Resource;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @author zhaoqiankun
 * @date 2021/11/9
 * spring方式用kafka
 */
@RestController
@RequestMapping(value = "/app/template")
public class SpringProducerTest {

    @Autowired(required = false)// 用来发送 kafka
    private KafkaTemplate kafkaTemplate;

    @PostMapping("/send")
    public  void send () {
        kafkaTemplate.send("demo","demo").addCallback(success->{
            System.out.println("发送成功");
        },error->{
            System.out.println("发送失败");
            error.printStackTrace();
        });

    }

    @PostMapping(value = "/kafka/testSend")
    public void sendMessage(String topic,String num) {
        for (int i =1;i<=num.length();i++) {
            String uuid = UUID.randomUUID().toString().replaceAll("-","");
            Device device = new Device(i ,uuid,false);
            String s = JSONObject.toJSONString(device);
            kafkaTemplate.send(topic,s).addCallback(success ->{
                System.out.println("发送成功");
            },fail->{
                System.out.println("发送失败"+s);
            });
        }

    }
}
